package com.example.hxds.snm.task;

import cn.hutool.core.map.MapUtil;
import com.example.hxds.common.exception.HxdsException;
import com.example.hxds.snm.entity.NewOrderMessage;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

@Component
@Slf4j
public class NewOrderMassageTask {

    @Autowired
    private ConnectionFactory factory;

    /**
     * 同步发送新订单消息
     */
    public void sendNewOrderMessage(ArrayList<NewOrderMessage> list){

        int ttl= 60 * 1000;//新订单缓存信息时间
        String exchangeName="new_order_private";
        try(
                Connection connection = factory.newConnection();
                final Channel channel = connection.createChannel()
        ) {
            //定义交换机，根据routeing key 路由消息
            channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
            HashMap param=new HashMap();
            for (NewOrderMessage message : list) {
                //MQ消息的属性信息
                HashMap map = new HashMap();
                map.put("orderId", message.getOrderId());
                map.put("from", message.getFrom());
                map.put("to", message.getTo());
                map.put("expectsFee", message.getExpectsFee());
                map.put("mileage", message.getMileage());
                map.put("minute", message.getMinute());
                map.put("distance", message.getDistance());
                map.put("favourFee", message.getFavourFee());

                //创建信息属性对象
                final AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
                        .expiration(ttl+"").contentEncoding("UTF-8").headers(map).build();

                String queueName="queue_" + message.getUserId(); //队列名字
                String routingKey=message.getUserId();
                //声明队列（持久化缓存消息，消息接收不加锁，消息全部接收完并不删除队列）
                channel.queueDeclare(queueName,true,false,false,param);
                //交换机绑定消息队列
                channel.queueBind(queueName,exchangeName,routingKey);
                //向交换机发送消息
                channel.basicPublish(exchangeName,routingKey,properties,("新订单"+message.getUserId()).getBytes());
                log.debug(message.getUserId()+"的新订单发送成功");

            }
        } catch (IOException | TimeoutException e) {
            log.error("执行异常",e);
            throw  new HxdsException("新订单发送失败");
        }
    }

    /**
     * 异步发送新订单消息
     */
    @Async
    public void sendNewOrderMessageAsync(ArrayList<NewOrderMessage> list) {
        sendNewOrderMessage(list);
    }

    /**
     * 同步接收订单消息
     */
    public ArrayList<NewOrderMessage> receiveNewOrderMessage(long userId ) {
        String exchangeName = "new_order_private"; //交换机名字
        String queueName = "queue_" + userId; //队列名字
        String routingKey = userId + ""; //routing key

        ArrayList<NewOrderMessage> list=new ArrayList<>();
        try(
                final Connection connection = factory.newConnection();
                final Channel channel = connection.createChannel()
        ){
            //定义交换机，routingKey模式
            channel.exchangeDeclare(exchangeName,BuiltinExchangeType.DIRECT);
            //声明队列（持久化缓存消息，消息接收不加锁，消息全部接收完并不删除队列）
            channel.queueDeclare(queueName,true,false,false,null);
            //绑定要接收的队列
            channel.queueBind(queueName,exchangeName,routingKey);
            //为了避免一次性接收太多消息，我们采用限流的方式，每次接收10条消息，然后循环接收
            channel.basicQos(10,false);

            while (true){
                //从队列中接收消息
                final GetResponse response = channel.basicGet(queueName, false);
                if (response!=null){
                    //消息属性对象
                    final AMQP.BasicProperties properties = response.getProps();
                    final Map<String, Object> map = properties.getHeaders();

                    String orderId = MapUtil.getStr(map, "orderId");
                    String from = MapUtil.getStr(map, "from");
                    String to = MapUtil.getStr(map, "to");
                    String expectsFee = MapUtil.getStr(map, "expectsFee");
                    String mileage = MapUtil.getStr(map, "mileage");
                    String minute = MapUtil.getStr(map, "minute");
                    String distance = MapUtil.getStr(map, "distance");
                    String favourFee = MapUtil.getStr(map, "favourFee");

                    //把新订单的消息封装到对象中
                    NewOrderMessage message = new NewOrderMessage();
                    message.setOrderId(orderId);
                    message.setFrom(from);
                    message.setTo(to);
                    message.setExpectsFee(expectsFee);
                    message.setMileage(mileage);
                    message.setMinute(minute);
                    message.setDistance(distance);
                    message.setFavourFee(favourFee);

                    list.add(message);
                    final byte[] body = response.getBody();
                    String msg=new String(body);
                    log.debug(msg);

                    //确认收到消息，MQ删除队列中的消息
                    final long deliveryTag = response.getEnvelope().getDeliveryTag();
                    channel.basicAck(deliveryTag,false);
                }else {
                    break;
                }
            }
        } catch(Exception e){
            log.error("执行异常",e);
            throw  new HxdsException("接收订单失败");
        }
        return list;
    }

    /**
     * 同步删除新订单消息队列
     */
    public void deleteNewOrderQueue(long userId) {
        String exchangeName = "new_order_private"; //交换机名字
        String queueName = "queue_" + userId; //队列名字
        try (Connection connection = factory.newConnection();
             Channel privateChannel = connection.createChannel();
        ) {
            //定义交换机
            privateChannel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
            //删除队列
            privateChannel.queueDelete(queueName);
            log.debug(userId + "的新订单消息队列成功删除");
        } catch (Exception e) {
            log.error(userId + "的新订单队列删除失败", e);
            throw new HxdsException("新订单队列删除失败");
        }
    }

    /**
     * 异步删除新订单消息队列
     */
    @Async
    public void deleteNewOrderQueueAsync(long userId) {
        deleteNewOrderQueue(userId);
    }

    /**
     * 同步清空新订单消息队列
     */
    public void clearNewOrderQueue(long userId) {
        String exchangeName =  "new_order_private";
        String queueName = "queue_" + userId;
        try (Connection connection = factory.newConnection();
             Channel privateChannel = connection.createChannel();
        ) {
            privateChannel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
            privateChannel.queuePurge(queueName);
            log.debug(userId + "的新订单消息队列清空删除");
        } catch (Exception e) {
            log.error(userId + "的新订单队列清空失败", e);
            throw new HxdsException("新订单队列清空失败");
        }
    }

    /**
     * 异步清空新订单消息队列
     */
    @Async
    public void clearNewOrderQueueAsync(long userId) {
        clearNewOrderQueue(userId);
    }
}
